package cn.iotab.iotkit.transport.api.access.device;

import cn.iotab.iotkit.transport.api.access.DefaultTransportAccessService;
import cn.iotab.iotkit.transport.api.access.TransportAccessConfig;
import cn.iotab.iotkit.transport.api.access.handler.DeviceMessageHandler;
import cn.iotab.iotkit.transport.api.access.handler.DeviceMessageSender;
import cn.iotab.iotkit.transport.api.access.handler.DeviceReplyListener;
import cn.iotab.iotkit.transport.api.access.handler.DeviceSessionHandler;
import cn.iotab.iotkit.transport.api.access.handler.DeviceSessionOps;
import cn.iotab.iotkit.transport.api.access.handler.ToServerRpcHandler;
import cn.iotab.iotkit.transport.api.access.handler.adaptor.ToServerMsgHandlerAdaptor;
import cn.iotab.iotkit.transport.api.access.handler.adaptor.ToServerRpcHandlerAdaptor;
import cn.iotab.iotkit.transport.api.access.handler.impl.DefaultDeviceMessageSender;
import cn.iotab.iotkit.transport.api.access.handler.impl.DefaultDeviceReplyListener;
import cn.iotab.iotkit.transport.api.access.handler.impl.DefaultDeviceSessionManager;
import cn.iotab.iotkit.transport.api.access.handler.impl.DefaultDeviceSessionOps;
import cn.iotab.iotkit.transport.api.common.queue.KafkaTransportQueueFactory;
import lombok.Getter;

import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 设备操作的入口
 */
public class DeviceManger {

    private final AtomicBoolean started = new AtomicBoolean(false);

    //配置信息
    private final TransportAccessConfig config;

    /**
     * 基础组件
     */
    private KafkaTransportQueueFactory queueFactory;
    private DefaultTransportAccessService transportAccessService;


    /**
     * 上行处理
     */
    private DeviceMessageHandler deviceMessageHandler;
    private DeviceReplyListener deviceReplyListener;
    private DeviceSessionHandler deviceSessionHandler;
    private ToServerRpcHandler toServerRpcHandler;
    /**
     * 下行处理
     */
    private DeviceMessageSender deviceMessageSender;
    private DeviceSessionOps deviceSessionOps;

    /**
     * 用户API
     */
    @Getter
    private DeviceOperator deviceOperator;



    public DeviceManger(TransportAccessConfig config) {
        this.config = config;
    }

    /**
     * 启动
     */
    public synchronized void init(HandlerManager handlerManager) {
        handlerManager.check();

        if (started.compareAndSet(false, true)) {
            /*
             * 基础组件初始化
             */
            queueFactory = new KafkaTransportQueueFactory(config.getQueueConfig());
            transportAccessService = new DefaultTransportAccessService(queueFactory, config);
            /*
             * 上下线消息处理
             */
            DefaultDeviceSessionManager sessionManager = new DefaultDeviceSessionManager();
            DefaultDeviceReplyListener defaultDeviceReplyListener = new DefaultDeviceReplyListener();
            DefaultDeviceSessionOps defaultDeviceSessionOps = new DefaultDeviceSessionOps(sessionManager, transportAccessService);

            deviceSessionHandler = sessionManager;
            deviceReplyListener = defaultDeviceReplyListener;

            deviceMessageSender = new DefaultDeviceMessageSender(transportAccessService, sessionManager, defaultDeviceReplyListener);
            deviceSessionOps = defaultDeviceSessionOps;

            /*
             * 用户调用设备的接口
             */
            deviceOperator = new DeviceOperator(deviceMessageSender, deviceSessionOps);
            /*
             * 接收设备消息的Handler
             */
            ToServerRpcHandlerAdaptor toServerRpcHandlerAdaptor = new ToServerRpcHandlerAdaptor(handlerManager);
            ToServerMsgHandlerAdaptor toServerMsgHandlerAdaptor = new ToServerMsgHandlerAdaptor(handlerManager,
                    deviceMessageSender, deviceReplyListener, deviceSessionHandler);
            transportAccessService.init(toServerRpcHandlerAdaptor, toServerMsgHandlerAdaptor);
        }
    }

    /**
     * 是否启动
     */
    public boolean isStarted() {
        return started.get();
    }
    /**
     * 销毁
     */
    public synchronized void destroy() {
        if (started.compareAndSet(true, false)) {
            //
            if (queueFactory != null){
                queueFactory.destroy();
            }
            //
            if (transportAccessService != null){
                transportAccessService.destroy();
            }
        }
    }
}
